package org.example.rabbitmq.client.hello;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class HelloWorld {
    private final static String QUEUE_NAME = "hello";

    public void sayHello(int times) throws IOException, TimeoutException {
        Connection connection = null;
        try {
            connection = newConnection();
            Channel channel = connection.createChannel();
            declareQueue(channel);

            for (int i = 0; i < times; i++) {
                String message = "Hello World! " + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
                System.out.println("[x] Sent message: " + message);
            }

            channel.basicRecover();
        } finally {
            closeConnection(connection);
        }
    }

    public void consumeMessage() throws InterruptedException, IOException, TimeoutException {
        Connection connection = null;
        CountDownLatch latch = new CountDownLatch(1);
        try {
            connection = newConnection();
            Channel channel = connection.createChannel();
            declareQueue(channel);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                        throws IOException {
                    try {
                        String message = new String(body, "UTF-8");
                        handleMessage(message);
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    } catch (Exception ex) {
                        channel.basicReject(envelope.getDeliveryTag(), true);
                        System.out.println("[x] Error: " + ex.getMessage());
                    }
                }
            };
            channel.basicConsume(QUEUE_NAME, false, consumer);

            System.out.println("[*] Waiting for consume messages.");
            latch.await(2, TimeUnit.SECONDS);
        } finally {
            closeConnection(connection);
        }
    }

    public void getMessage() throws IOException, TimeoutException {
        Connection connection = null;
        try {
            connection = newConnection();
            Channel channel = connection.createChannel();
            declareQueue(channel);
            System.out.println("[*] Waiting for get messages.");

            GetResponse response = channel.basicGet(QUEUE_NAME, false);
            if (response != null) {
                byte[] body = response.getBody();
                String message = new String(body, "UTF-8");
                System.out.println("[x] Get message: " + message);
                System.out.println("[x] Remaining message count: " + response.getMessageCount());
                channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
            }
        } finally {
            closeConnection(connection);
        }
    }

    private void handleMessage(String message) {
        System.out.print("[x] Consume message: ");
        for (char c : message.toCharArray()) {
            System.out.print(c);
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
            }
        }
        System.out.println();
    }

    private Connection newConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest@local");
        return factory.newConnection();
    }

    private void declareQueue(Channel channel) throws IOException {
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    }

    private void closeConnection(Connection connection) {
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

}
